/*=====================================================================*
 *                   Copyright (C) 2011 Paul Mineiro                   *
 * All rights reserved.                                                *
 *                                                                     *
 * Redistribution and use in source and binary forms, with             *
 * or without modification, are permitted provided that the            *
 * following conditions are met:                                       *
 *                                                                     *
 *     * Redistributions of source code must retain the                *
 *     above copyright notice, this list of conditions and             *
 *     the following disclaimer.                                       *
 *                                                                     *
 *     * Redistributions in binary form must reproduce the             *
 *     above copyright notice, this list of conditions and             *
 *     the following disclaimer in the documentation and/or            *
 *     other materials provided with the distribution.                 *
 *                                                                     *
 *     * Neither the name of Paul Mineiro nor the names                *
 *     of other contributors may be used to endorse or promote         *
 *     products derived from this software without specific            *
 *     prior written permission.                                       *
 *                                                                     *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND              *
 * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,         *
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES               *
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE             *
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER               *
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,                 *
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES            *
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE           *
 * GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR                *
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF          *
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT           *
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY              *
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE             *
 * POSSIBILITY OF SUCH DAMAGE.                                         *
 *                                                                     *
 * Contact: Paul Mineiro <paul@mineiro.com>                            *
 *=====================================================================*/

#ifndef __THREAD_QUEUE_HH_
#define __THREAD_QUEUE_HH_

#include <boost/thread.hpp>
#include <queue>

#include "example.hh"

namespace flassol
{
  template<typename Data>
  class ThreadQueue
    {
      private:
        std::queue<Data> queue;
        boost::mutex mutex;
        boost::condition_variable non_empty;
        boost::condition_variable not_full;
        volatile unsigned int max_length;
        volatile unsigned int num_writers;
        volatile bool first_add_writer;

      public:
        typedef Data data;

        ThreadQueue (unsigned int _max_length) : 
          max_length (_max_length),
          num_writers (1),
          first_add_writer (true)
            {
            }
        
        void
        push (const Data& data)
          {
            boost::mutex::scoped_lock lock (mutex);

            while (queue.size () >= max_length)
              {
                not_full.wait (lock);
              }

            bool was_empty = queue.empty ();

            queue.push (data);

            lock.unlock ();

            if (was_empty)
              {
                non_empty.notify_one ();
              }
          }

        bool
        pop (Data& data)
          {
            boost::mutex::scoped_lock lock (mutex);

            while (queue.empty () && num_writers)
              {
                non_empty.wait (lock);
              }

            if (queue.empty ())
              {
                lock.unlock ();

                return false;
              }

            bool was_full = (queue.size () >= max_length);

            data = queue.front ();
            queue.pop ();

            lock.unlock ();

            if (was_full)
              {
                not_full.notify_one ();
              }

            return true;
          }

        void
        add_writer ()
          {
            boost::mutex::scoped_lock lock (mutex);
            if (! first_add_writer)
              {
                ++num_writers;
              }
            first_add_writer = false;
            lock.unlock ();
          }

        void
        del_writer ()
          {
            bool last_writer;

            boost::mutex::scoped_lock lock (mutex);
            --num_writers;
            last_writer = (num_writers == 0);
            lock.unlock ();

            if (last_writer)
              {
                non_empty.notify_all ();
              }
          }
    };

  template<typename Queue>
  class Tee
    {
      private:
        std::vector<Queue*> out;

      public:
        Tee (std::vector<Queue*> _out)
          : out (_out)
          {
          }

        void
        push (const typename Queue::data& data)
          {
            for (unsigned int i = 0; i < out.size (); ++i)
              {
                out[i]->push (data);
              }
          }

        void
        add_writer ()
          {
            for (unsigned int i = 0; i < out.size (); ++i)
              {
                out[i]->add_writer ();
              }
          }

        void
        del_writer ()
          {
            for (unsigned int i = 0; i < out.size (); ++i)
              {
                out[i]->del_writer ();
              }
          }

        size_t
        num_consumers () const
          {
            return out.size ();
          }
    };

  typedef ThreadQueue<std::vector<boost::optional<RegressionExample> >* > RegressionQueue;
  typedef ThreadQueue<std::vector<boost::optional<CSMCExample> >* > CSMCQueue;
};

#endif // __THREAD_QUEUE_HH_
